package com.qch.kafka.demo.serialization;

import com.qch.kafka.demo.entity.User;
import org.apache.kafka.common.errors.SerializationException;
import org.apache.kafka.common.serialization.Serializer;

import java.io.UnsupportedEncodingException;
import java.nio.ByteBuffer;
import java.util.Map;

/**
 * @Author qch
 * @Date 2021/5/31
 */
public class Userserializer implements Serializer<User> {
    public void configure(Map<String, ?> map, boolean b) {
        //用于接收对序列化的配置参数，并对当前序列化器进行配置和初始化
    }

    public byte[] serialize(String topic, User data) {
        try {
        if(data==null)
        {
            return null;
        }else {
            Integer userId = data.getUserId();
            String name = data.getName();
            if(userId!=null){
                if(name!=null)
                {

                    byte[] bytes = name.getBytes("UTF-8");
                    int length=bytes.length;
                    /**
                     * 第一个4表示：Integer类型的userId需要+4（4个字节）
                     * 第二个4表示：用于存储userName字节数组的长度int值
                     * 第三个长度，用于存放username序列化之后的字节数组
                     */
                    ByteBuffer buffer = ByteBuffer.allocate(4+4+length);
                    buffer.putInt(userId);
                    buffer.putInt(length);
                    //设置username字节数组
                    buffer.put(bytes);
                    //以字节数组形式返回user对象的值
                    return buffer.array();

                }
            }
        }   } catch (Exception e) {
          throw new SerializationException("序列化失败");
        }
        return  null;
    }

    public void close() {
                //用于关闭资源等操作，需要幂等，即多次调用，效果是一样的

    }
}
